package com.lagou.registry.listener;

import com.alibaba.fastjson.JSON;
import com.lagou.consumer.ConsumerNettyConnect;
import com.lagou.registry.RpcProviderNode;
import com.lagou.registry.ServerNode;
import com.lagou.registry.ZkClientHolder;
import com.lagou.registry.ZkProviderRegistry;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;

/**
 * 负责监听生产者的注册路径节点信息
 */
public class ConsumerListener implements Runnable {


    private ConsumerNettyConnect nettyConnect;

    private String watchPath;

    private CuratorFramework client;

    private ZkProviderRegistry providerRegistry;

    public ConsumerListener(String watchPath,ConsumerNettyConnect conn,ZkProviderRegistry providerRegistry){
        this.client = ZkClientHolder.client;
        this.watchPath = watchPath;
        this.nettyConnect = conn;
        this.providerRegistry = providerRegistry;
    }

    @Override
    public void run() {
        try {
            //创建PathChildrenCache
            //参数：true代表缓存数据到本地
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client,watchPath,true);
            //BUILD_INITIAL_CACHE 代表使用同步的方式进行缓存初始化。
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            pathChildrenCache.getListenable().addListener((cf, event) -> {
                PathChildrenCacheEvent.Type eventType = event.getType();
                String data = new String(event.getData().getData());
                ServerNode serverNode = JSON.parseObject(data, RpcProviderNode.class);
                String ip = serverNode.getIp();
                Integer port = serverNode.getPort();
                switch (eventType) {
                    case CONNECTION_RECONNECTED:
                        for (ChildData currentDatum : pathChildrenCache.getCurrentData()) {
                            String s = new String(currentDatum.getData());
                            System.out.println("1缓存数据？？"+s);
                        }
                        pathChildrenCache.rebuild();
                        nettyConnect.conn(ip,port);
                        System.out.println("Connection reconnected");
                        break;
                    case CONNECTION_SUSPENDED:
                        break;
                    case CONNECTION_LOST:
                        System.out.println("Connection lost");
                        break;
                    case CHILD_ADDED:
                        nettyConnect.conn(ip,port);
                        System.out.println("ip:"+ip+",port:"+port+"netty服务器已经加入");
                        break;
                    case CHILD_UPDATED:
                        System.out.println("Child updated");
                        break;
                    case CHILD_REMOVED:
                        //bootstrap.group().shutdown();
                        //nettyConnect.disconnect(ip,port);
                        System.out.println("!!!!!!!!!!ip:"+ip+",port:"+port+"netty服务器已经shutdown");
                        break;
                    default:
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
